package org.basis.nio.c4;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import static org.basis.nio.c2.ByteBufferUtil.debugAll;

/**
 * 多线程改进server端
 *
 * @author Mr_wenpan@163.com 2021/09/15 16:39
 */
@Slf4j
public class MultiThreadServer {

    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        Selector boss = Selector.open();
        SelectionKey bossKey = ssc.register(boss, 0, null);
        bossKey.interestOps(SelectionKey.OP_ACCEPT);
        ssc.bind(new InetSocketAddress(8080));

        log.info("server启动并绑定到端口并注册到selector上完毕.......");

        // 1. 创建固定数量的(和操作系统核心数一样) worker 并初始化
        Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Worker("worker-" + i);
        }
        AtomicInteger index = new AtomicInteger();

        while (true) {
            // boss 在这里阻塞等待客户端连接
            boss.select();
            Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                iter.remove();
                ServerSocketChannel scc = (ServerSocketChannel) key.channel();
                SocketChannel channel = scc.accept();
                // 注册到对应的worker
                channel.configureBlocking(false);
                log.info("connected.....{}", channel.getRemoteAddress());
                log.info("before register.....{}", channel.getRemoteAddress());
                // round robin 轮询
                workers[index.getAndIncrement() % workers.length].register(channel);
                log.info("after register.....{}", channel.getRemoteAddress());
            }
        }

    }

    static class Worker implements Runnable {

        private Thread thread;
        private Selector selector;
        private String name;
        // 该worker是否已经启动
        private volatile boolean start = false;
        // 队列
        private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();

        public Worker(String name) {
            this.name = name;
        }

        /**
         * 为什么要将这些属性初始化的动作放在这里呢？为什么不放在构造函数？
         * 因为懒加载，当没有channel来注册的时候这个worker的线程就不会启动
         */
        public void register(SocketChannel sc) throws IOException {
            // 犹豫模式，如果worker还没有启动则启动他
            if (!start) {
                synchronized (this) {
                    if (!start) {
                        selector = Selector.open();
                        // 将本对象的run方法传递给thread
                        thread = new Thread(this, name);
                        thread.start();
                        start = true;
                    }
                }
            }
            // 方式一、队列的方式实现
//            queue.add(() -> {
//                // 注册channel到selector上
//                try {
//                    sc.register(selector, SelectionKey.OP_READ, null);
//                } catch (ClosedChannelException e) {
//                    e.printStackTrace();
//                }
//            });
            // 唤醒selector，并且将channel注册到该selector上（这里的唤醒类似于park和unpark，仔细理解）
            selector.wakeup();
            // 方式二、唤醒后直接注册
            sc.register(selector, SelectionKey.OP_READ, null);
        }

        @Override
        public void run() {

            while (true) {
                try {
                    // selector阻塞等待事件的发生[当上面调用了selector.wakeup();方法后这里阻塞会被中断]
                    selector.select();

                    // 如果队列中有等待添加的channel，则将这些channel获取出来添加到selector中
                    // queue.peek()方法非阻塞，如果没有元素，则返回空
//                    while (queue.peek() != null) {
//                        queue.poll().run();
//                    }

                    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();

                    // 因为这里要一边遍历一边删除所以需要使用迭代器
                    while (iter.hasNext()) {
                        SelectionKey key = iter.next();
                        iter.remove();
                        if (key.isReadable()) {
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            SocketChannel channel = (SocketChannel) key.channel();
                            log.info("read......{}", channel.getRemoteAddress());
                            channel.read(buffer);
                            buffer.flip();
                            // 打印接收到的信息
                            debugAll(buffer);
                        }
                    }

                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}


